package com.pangyangyang.elasticsearch.plugin.collector;

import com.pangyangyang.elasticsearch.plugin.exception.MessageRiverException;
import com.pangyangyang.elasticsearch.plugin.river.Build;
import com.pangyangyang.elasticsearch.plugin.river.MessageRiver;
import com.pangyangyang.elasticsearch.plugin.river.MessageRiverSettings;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;
import org.elasticsearch.common.settings.Settings;

import java.lang.reflect.Constructor;
import java.util.Map;

/**
 * 负责创建Collector，必要时考虑这里注入
 * Created by yangwang on 2015/1/15.
 */
public class EventCollectorFactory {

    private static ESLogger logger = ESLoggerFactory.getLogger(Build.LOGGER_NAME);

    public static  <T  extends EventCollector> T  createAndStartCollector(MessageRiverSettings settings, String collectorName){
        T eventCollector;
        Class<T> collectorClass = (Class<T>)Build.getInstance().getCollectorClasses().get(collectorName);
        if (collectorClass == null){
            throw new MessageRiverException(String.format("collector not found %s.", collectorName));
        }

        try {
            Constructor<T> constructor = collectorClass.getConstructor(MessageRiverSettings.class);
            eventCollector = constructor.newInstance(settings);
        } catch (Exception e) {
            throw new MessageRiverException(String.format("create collector failed %s.", collectorClass.getName()), e);
        }

        return  eventCollector;
    }

}
